package each_webcode

import common.HbaseTools
import org.apache.spark.broadcast.Broadcast
import org.apache.spark.rdd.RDD
import org.apache.spark.{SparkConf, SparkContext}

import scala.collection.mutable
import scala.collection.mutable.HashMap
import scala.io.Source
import scala.util.parsing.json.JSON;
object transferValueCode_to_Name_fao {
        def main(args: Array[String]): Unit = {
                val sparkConf = new SparkConf().setAppName("FAOdatachange")//.setMaster("local[*]")
                val sc = new SparkContext(sparkConf)
                val baseInfotable: String = new String("BaseMapInfos") //DetailMapInfos
                val detailTable: String = new String("DetailMapInfos") //BaseMapInfos

                val inpath = args(0) //"hdfs://slave1:8020//InternationalData/IMF_Value.json"
                val outpath = args(1) //"hdfs://slave1:8020//InternationalData_change_after/imf_change.value"
                val numspartion = args(2)

                //val numspartion = "2"
                        //val inpath = "D:\\InternationalData\\makeChange\\src\\main\\resources\\sub.Fao_Value.json"
                        //val outpath = "hdfs://node01:8020//tmp/InternationalData_tmp/sub_fao_test_15"
                        val webcodes = List("FAO")
                //make a cache:
                        val inputFile = inpath //读取json文件
                val jsonStr = sc.textFile(inputFile) //.sample(true,0.01);

                val result: RDD[Option[Any]] = jsonStr.map(s => JSON.parseFull(s))

                val rdds = result.map {
                        r =>
                                r match {
                                        case Some(maps: Map[String, Any]) => {
                                                val webcode = "FAO"
                                                val thashmap: HashMap[String, String] = new HashMap[String, String]()
                                                maps.foreach(t => {
                                                        thashmap += (t._1.toString -> t._2.toString.replace("\n",""))
                                                        })
                                                //maybe there should deal with _id
                                                thashmap("_id") = thashmap("_id").substring(12, 36)
                                                // get webcode + sourcecode
                                                // rowkeys
                                                //处理 sourcecode
                                                thashmap("SourceCode") = thashmap("SourceCode")
                                                //[0,80)
                                                //map(随机数;FAO;SourceCode,json转化成为map数组)
                                                (scala.util.Random.nextInt(80).toString + ";" + webcode + ";" + thashmap("SourceCode"), thashmap)
                                                //("0"+"_WTO"+"_"+thashmap("SourceCode"),thashmap)
                                                }
                                        }
                        }
                //将迭代器中的每条数据变成一个string
                        val res:RDD[String]=rdds.groupByKey(numspartion.toInt).map { //numspartion = args(2)
                        case (k, v) => {
                                val k1array = k.split(";")
                                //FAO_SourceCode
                                        val k1 = k1array(1) + "_" + k1array(2)
                                //FAO_
                                        val pk1 = k1array(1) + "_"
                                //by k,you will get the mapinfos
                                        val tools = new HbaseTools()
                                val connection = tools.getConnection()
                                //得到dealtype对应的字段
                                        //返回表名:"BaseMapInfos",row_key:FAO_SourceCode,列族"dealTypes",map(kbase表的列名,列对应type类型)-------------------来自make_map01_map02_to_hb
                                        val dealTypes: Map[String, String] =  tools.getFamily_column_value(connection,baseInfotable, k1, "dealTypes").toMap
                                //得到map01对应的字段
                                        //返回表名:"BaseMapInfos",row_key:FAO_SourceCode,列族"map01",map(kbase列名,value表列名)-------------------来自make_map01_map02_to_hb
                                        val map01: Map[String, String] = tools.getFamily_column_value(connection,baseInfotable, k1, "map01").toMap
                                //得到map02对应的字段
                                        //返回表名:"BaseMapInfos",row_key:FAO_SourceCode,列族"map02",map(value表的列名,concept表的conceptId)-------------------来自make_map01_map02_to_hb
                                        val map02: Map[String, String] = tools.getFamily_column_value(connection,baseInfotable, k1, "map02").toMap
                                //create a map for map04
                                //then for each v;you can transfer code to value

                                //map(conceptId,map(detailId,DetailName))
                                        //-----------------------------------正常来说这个代码中需要的是列族WebInfo,但是这张表里没有这个列族,这个webinfo是由代码makeBaseMap01_step01写进来的
                                        //get "DetailMapInfos","FAO_TP"
                                        //get "DetailMapInfos","FAO_TM"
                                        //存放的是表名"DetailMapInfos",row_key:FAO_SourceCode对应的每个列族的列名和列值组成的map
                                        //get "DetailMapInfos","FAO_TP"
                                        //列族       列名                                              列值
                                        //conceptId:detailId                                      DetailName
                                        //例子:Area:110                timestamp=1608597314069, value=Japan
                                        //     Item:76                 timestamp=1608597314399, value=Oats rolled
                                        //共存了四个列族:Area,Item,Element,Year
                                        //------------------------------------是从makeDetailDic_step03
                                        val map04: HashMap[String, Map[String, String]] = new HashMap[String, Map[String, String]]()
                                //FAO 下任一SourceCode的 DetailMapInfos
                                        //得到表"DetailMapInfos",row_key:FAO_SourceCode对应的所有列族组成的Array
                                        val selffamilys = tools.getFamilys_list(connection,detailTable, k1)
                                //map04为map(列族,map(列名,列值))
                                for (onefamily: String <- selffamilys) {
                                        //得到表"DetailMapInfos",row_key:FAO_SourceCode对应的每个列族的列名和列值组成的map
                                                val tmp = tools.getFamily_column_value(connection,detailTable, k1, onefamily).toMap
                                        map04.put(onefamily, tmp)
                                        }

                                //map05
                                        //存放的是表名"DetailMapInfos",row_key:FAO_对应的列族("Unit Name","Flag")的列名和列值组成的map
                                        //get "DetailMapInfos","FAO_"-------------------------------------找不到列族"Unit Name","Flag"的数据是从哪里写进来的
                                        //列族       列名                                                列值
                                        //conceptId:detailId                                           DetailName
                                        //例子Unit Name:kg/TJ         timestamp=1608597314187, value=kilograms per terajoule
                                        //例子Unit Name:kg/ha         timestamp=1608597314069, value=kilograms per hectare
                                        //map05 FAO_ 公用字典对应关系：Flag ，Unit Name
                                        val map05: HashMap[String, Map[String, String]] = new HashMap[String, Map[String, String]]()
                                //FAO 下任一SourceCode的 DetailMapInfos
                                        val publicfamilys = Array("Unit Name","Flag")
                                //"Unit Name"列族中存的是单位名称的多个列和列值
                                //"Flag"列族存的是
                                for (onefamily: String <- publicfamilys) {
                                        val tmp = tools.getFamily_column_value(connection,detailTable, pk1, onefamily)
                                        map05.put(onefamily, tmp.toMap)
                                        }

                                //得到表名为"BaseMapInfos",rowkey为"public",列族为"public_Time"的所有列名和列值组成的map(detailname,detailId)---------------------------来自代码make_Dic
                                        //getFamily_column_value(connection:Connection,tableName: String, rowkey: String, columnFamily: String)
                                        val PU_timeMap:Map[String,String] = tools.getFamily_column_value(connection,baseInfotable,"public","public_Time").toMap
                                //得到表名为"BaseMapInfos",rowkey为"public",列族为public_country"的所有列名和列值组成的map(detailname,detailId)-------------------------来自代码make_Dic
                                        val PU_country:Map[String,String] = tools.getFamily_column_value(connection,baseInfotable,"public","public_country").toMap
                                //special
                                connection.close()
                                //key为webcoede;source,value为map(json的列,值)组成的迭代器
                                        //迭代器中的每一条数据返回一个string,以为一个迭代器中的数据返回Iterable[String]
                                        //再将迭代器中的每条数据变成一个string
                                        val onekeysValue:Iterable[String] = v.map {
                                        detailValuecode => { // 迭代器中的每一条
                                                var oneLinceMap:HashMap[String,String] = new mutable.HashMap[String,String]()
                                                //map03:map(Value表列名,Value表列值)
                                                        var map03 = detailValuecode
                                                //dealTypes:map(kbase表的列名,列对应type类型)
                                                //遍历kbase表的列名
                                                dealTypes.keys.foreach {
                                                        keys =>
                                                                dealTypes(keys) match {//kbase表的列名,对应的type类型
                                                                        case "type01" => {
                                                                                //map01 + map03
                                                                                //map01:map(kbase列名,value表列名),map03:map(Value表列名,Value表列值)
                                                                                if (map01.contains(keys) && map03.contains(map01(keys)) && map03(map01(keys))!="NA") {
                                                                                        oneLinceMap.put(keys,map03(map01(keys)))
                                                                                        }
                                                                                }
                                                                                case "type02" => {
                                                                                        //map01 + map03 + map04
                                                                                        //map01:map(kbase列名,value表列名),map04:map(conceptId,map(detailId,DetailName))
                                                                                        //map(kbase表的列名,map())
                                                                                        oneLinceMap.put(keys,map04("WebInfo")(map01(keys)))
                                                                                        }
                                                                                case "type03" => {
                                                                                        /*map01 + map02 =conceptid
                                                                                       map01 + map03 =detailId
                                                                                       conceptid + detailid + map04
                                                                                      */
                                                                                        //map01:map(kbase列名,value表列名)
                                                                                        //map02:map(value表的列名,concept表的conceptId)
                                                                                        //map03:map(Value表列名,Value表列值)
                                                                                        //map04:map(conceptId,map(detailId,DetailName))
                                                                                        if (map01.contains(keys) && map03.contains(map01(keys)) && map02.contains(map01(keys)) && map04.contains(map02(map01(keys))) && map04(map02(map01(keys))).contains(map03(map01(keys)))&&map03(map01(keys))!="NA") {
                                                                                                oneLinceMap.put(keys,map04(map02(map01(keys)))(map03(map01(keys))))
                                                                                                }

                                                                                        }
                                                                                case "type04" => {
                                                                                        //map01 + map02 = conceptId
                                                                                        oneLinceMap.put(keys,map02(map01(keys)))
                                                                                        }
                                                                                //map01:map(kbase列名,value表列名)
                                                                                //map02:map(value表的列名,concept表的conceptId)
                                                                                //map03:map(Value表列名,Value表列值)
                                                                                //map04:map(conceptId,map(detailId,DetailName))
                                                                                case "type05" => {
                                                                                        //map01 + map03 = detailId  map04(map01(k))(map03(map01(k)))
                                                                                        //one_column_value += ((keys+"\t"+map03(map01(keys)))+"(<>)")
                                                                                        if (map01.contains(keys) && map03.contains(map01(keys)) && map04.contains(map01(keys)) && map04(map01(keys)).contains(map03(map01(keys)))) {
                                                                                                oneLinceMap.put(keys,map04(map01(keys))(map03(map01(keys))))
                                                                                                }
                                                                                        }
                                                                                case "type06" => {
                                                                                        // map01 , a changeless value
                                                                                        if(map03.contains(map01(keys))){
                                                                                                oneLinceMap.put(keys,map01(keys))
                                                                                                }
                                                                                        }
                                                                                case "type08" => {
                                                                                        //map01 +  map02   get
                                                                                                // COUNTRY;DATASOURCE;DHSMICSGEOREGION;GBDREGION;MGHEREG;REGION;UNREGION;UNSDGREGION;WHOINCONMEREGION;WORLDBANKINCOMEGROUP
                                                                                                //+ map04 get dic
                                                                                                //map01 + map03 get code
                                                                                                val code = map03(map01(keys))
                                                                                        val dics = map02(map01(keys)).split(";")
                                                                                        var status = true
                                                                                        for (dickey <- dics) {
                                                                                                if (status && map04(dickey).contains(code)) {
                                                                                                        oneLinceMap.put(keys,map04(dickey)(code))
                                                                                                        status = false
                                                                                                        }
                                                                                                }
                                                                                        }
                                                                                case "type07" => {
                                                                                        // conceptid and detailid from differ column
                                                                                        //
                                                                                        if (map01.contains(keys) && map03.contains(map01(keys)) && map04.contains(map03(map01(keys)))) {
                                                                                                oneLinceMap.put(keys,map04(map03(map01(keys)))("ConceptName"))
                                                                                                }
                                                                                        }
                                                                                case "type09" => {
                                                                                        //map04(map01(k))("ConceptName")
                                                                                        if (map01.contains(keys) && map04.contains(map01(keys)) && map04(map01(keys)).contains("ConceptName")) {
                                                                                                oneLinceMap.put(keys,map04(map01(keys))("ConceptName"))
                                                                                                }
                                                                                        }
                                                                                case "type01_fao" =>{
                                                                                        //type01_fao FootNote Unit;Note
                                                                                        //map05("Unit Name")(map03("Unit"))
                                                                                        // map03("Note")
                                                                                        if(map03.contains("Note") && map03.contains("Unit") && map05.contains("Unit Name") && map05("Unit Name").contains(map03("Unit"))){
                                                                                                oneLinceMap.put(keys,map05("Unit Name")(map03("Unit"))+"，"+map03("Note"))
                                                                                                }else if(map03.contains("Unit") && map05.contains("Unit Name") && map05("Unit Name").contains(map03("Unit"))){
                                                                                                oneLinceMap.put(keys,map05("Unit Name")(map03("Unit")))
                                                                                                }
                                                                                                else{

                                                                                                }
                                                                                        }
                                                                                case "type02_fao" =>{
                                                                                        //type02_fao Reporter Survey_Code Survey
                                                                                        if(map03.contains("Survey_Code") && map04.contains("Survey") && map04("Survey").contains(map03("Survey_Code"))){
                                                                                                val infos:Array[String] = map04("Survey")(map03("Survey_Code")).split("-")
                                                                                                if(infos.length >= 2) {
                                                                                                        oneLinceMap.put("Reporter",infos(0))
                                                                                                        if(infos.length>2){
                                                                                                                oneLinceMap.put("Year",infos(1).toString+"-"+infos(2).toString)
                                                                                                                }else{

                                                                                                                oneLinceMap.put("Year",infos(1))
                                                                                                                }
                                                                                                        }else{

                                                                                                        }
                                                                                                }
                                                                                        }
                                                                                case "type03_fao" =>{
                                                                                        if(map03.contains("Item_Code") && map03.contains("CurrencyCode")){
                                                                                                oneLinceMap.put("IndicatorCategory",map03("Item_Code")+"-"+map03("CurrencyCode"))
                                                                                                }
                                                                                        }
                                                                                case "type04_fao" =>{
                                                                                        if(map04.contains("Item") &&map04.contains("Currency") && map04("Item").contains(map03("Item_Code")) && map04("Currency").contains(map03("CurrencyCode"))){
                                                                                                oneLinceMap.put("Indicator",map04("Item")(map03("Item_Code"))+"-"+map04("Currency")(map03("CurrencyCode")))
                                                                                                }
                                                                                        }
                                                                                case "type05_fao" =>{
                                                                                        if(map03.contains("Months_Code") && map03("Months_Code")!="" &&map03("Months_Code")!="NA"){
                                                                                                oneLinceMap.put("Period",map03("Months_Code")(0).toString)
                                                                                                }

                                                                                        }
                                                                                case "type06_fao" =>{
                                                                                        if(map03.contains("Indicator") && map03.contains("SeriesDescription")){
                                                                                                oneLinceMap.put("Indicator",map03("Indicator") + ":"+map03("SeriesDescription"))
                                                                                                }
                                                                                        }
                                                                                case "type07_fao" =>{
                                                                                        if(map03.contains("LowerBound") && map03.contains("UpperBound")){
                                                                                                if(map03("LowerBound")!="NA"&&map03("UpperBound")!="NA"){
                                                                                                        oneLinceMap.put("ValueRange" ,map03("LowerBound") + "-"+map03("UpperBound"))
                                                                                                        }else if(map03("LowerBound")!="NA"&&map03("UpperBound")=="NA"){
                                                                                                        oneLinceMap.put("ValueRange" ,map03("LowerBound"))
                                                                                                        }
                                                                                                        else if(map03("LowerBound")=="NA"&&map03("UpperBound")!="NA"){
                                                                                                        oneLinceMap.put("ValueRange",map03("UpperBound"))
                                                                                                        }
                                                                                                }
                                                                                        }
                                                                                case "type08_fao" =>{
                                                                                        if(map03.contains("Nature") && map03.contains("OBS_STATUS") && map04.contains(map02("Nature"))&&map04.contains(map02("OBS_STATUS"))
                                                                                                &&  map04(map02("Nature")).contains(map03("Nature")) && map04(map02("OBS_STATUS")).contains(map03("OBS_STATUS"))
                                                                                                ){
                                                                                                if(map03("Nature")!="NA" && map03("OBS_STATUS")!="NA")
                                                                                                        {
                                                                                                                oneLinceMap.put("ObservationStatus",map04(map02("Nature"))(map03("Nature"))+"-"+map04(map02("OBS_STATUS"))(map03("OBS_STATUS")))

                                                                                                                }else if(map03("Nature")=="NA" && map03("OBS_STATUS")!="NA"){
                                                                                                        oneLinceMap.put("ObservationStatus",map04(map02("OBS_STATUS"))(map03("OBS_STATUS")))
                                                                                                        }else if(map03("Nature")!="NA" && map03("OBS_STATUS")=="NA"){
                                                                                                        oneLinceMap.put("ObservationStatus",map04(map02("Nature"))(map03("Nature")))
                                                                                                        }
                                                                                                }
                                                                                        }
                                                                                case "type09_fao" =>{
                                                                                        if(map03.contains("Flag") && map05.contains(map03("Flag"))){
                                                                                                oneLinceMap.put("ObservationStatus",map05("Flag")(map03("Flag")))
                                                                                                }
                                                                                        }
                                                                        }
                                                        }
                                                //合并字段，生成
                                                //检测 Year字段，
                                                if(oneLinceMap.contains("Year")&&oneLinceMap("Year")!=""){
                                                        if(oneLinceMap("Year").length>4 && !oneLinceMap.contains("Period")){
                                                                oneLinceMap.put("Period",oneLinceMap("Year").substring(4))
                                                                oneLinceMap("Year") = oneLinceMap("Year").substring(0,4)
                                                                }
                                                        //检测Period字段
                                                        //1检查是否有值，
                                                        }

                                                //生成TimeSort字段；Period字段判断是否存在并进行转换，
                                                if(oneLinceMap.contains("Year")&&oneLinceMap.contains("Period")&&oneLinceMap("Period").trim!=""){
                                                        if(PU_timeMap.contains(oneLinceMap("Period").trim)){
                                                                oneLinceMap.put("TimeSort",oneLinceMap("Year")+PU_timeMap(oneLinceMap("Period").trim))
                                                                }else{
                                                                oneLinceMap.put("TimeSort",oneLinceMap.getOrElse("Year",""))
                                                                }
                                                        }else{
                                                        oneLinceMap.put("TimeSort",oneLinceMap.getOrElse("Year",""))
                                                        }
                                                //生成FuzzyIndicator字段
                                                if(oneLinceMap.contains("Indicator")&&oneLinceMap("Indicator")!="")
                                                        {oneLinceMap.put("FuzzyIndicator",oneLinceMap("Indicator"))
                                                                //生成IndicatorLength字段
                                                                oneLinceMap.put("IndicatorLength",oneLinceMap("Indicator").length.toString)
                                                                }
                                                //生成FuzzyReporter字段
                                                if(oneLinceMap.contains("Reporter")&&oneLinceMap("Reporter")!=""){
                                                        if(PU_country.contains(oneLinceMap("Reporter"))){
                                                                oneLinceMap.put("FuzzyReporter",oneLinceMap("Reporter")+"，"+PU_country(oneLinceMap("Reporter")))
                                                                }else{
                                                                oneLinceMap.put("FuzzyReporter",oneLinceMap("Reporter"))
                                                                }
                                                        }
                                                //Parterner
                                                //生成FUZZYPARTERNER字段
                                                if(oneLinceMap.contains("Parterner")&&oneLinceMap("Parterner")!=""){
                                                        if(PU_country.contains(oneLinceMap("Parterner"))){
                                                                oneLinceMap.put("FUZZYPARTERNER",oneLinceMap("Parterner")+"，"+PU_country(oneLinceMap("Parterner")))
                                                                }else{
                                                                oneLinceMap.put("FUZZYPARTERNER",oneLinceMap("Parterner"))
                                                                }
                                                        }
                                                val MetaData:Array[String] = Array("WebCode", "SourceCode", "ReporterCode", "Reporter", "ParternerCode", "Parterner", "Year", "Period",
                                                        "IndicatorCategory", "Indicator", "Dim1Type", "Dim1", "Dim2Type", "Dim2", "Dim3Type", "Dim3",
                                                        "Dim4Type", "Dim4", "Dim5Type", "Dim5", "Dim6Type", "Dim6", "Dim7Type", "Dim7", "Dim8Type", "Dim8" )
                                                val DIMTYPEMETADATA:Array[String] =Array("Dim1Type", "Dim2Type", "Dim3Type", "Dim4Type", "Dim5Type", "Dim6Type", "Dim7Type", "Dim8Type")
                                                val DIMMETADATA:Array[String] = Array("Dim1", "Dim2", "Dim3", "Dim4", "Dim5", "Dim6", "Dim7", "Dim8")

                                                //生成DIMTYPEMETADATA
                                                        var dimtypemetadata = DIMTYPEMETADATA.flatMap(oneLinceMap.getOrElse(_,"<->")+"，").mkString("").replace("<->，","")

                                                //生成DIMMETADATA
                                                        var dimemetadata = DIMMETADATA.flatMap(
                                                        onecel=>{
                                                                oneLinceMap.getOrElse(onecel,"<->")+"，"
                                                                }
                                                        ).mkString("").replace("<->，","")
                                                //生成MetaData
                                                        var metadata = MetaData.flatMap(oneLinceMap.getOrElse(_,"<->")+"，").mkString("").replace("<->，","")
                                                oneLinceMap.put("DIMTYPEMETADATA",dimtypemetadata)
                                                oneLinceMap.put("DIMMETADATA",dimemetadata)
                                                oneLinceMap.put("MetaData",metadata)
                                                //val kkl:RDD[String] = sc.parallelize(().toString)
                                                oneLinceMap.mkString("(<>)").replace(" -> ","\t")
                                                //val kkl:RDD[String] = sc.parallelize(().toString)
                                                //val kkl:RDD[String] = sc.parallelize(().toString)
                                                }
                                        }
                                var fi = onekeysValue.mkString( "\n")
                                fi
                                }
                        }
                res.saveAsTextFile(outpath)
                sc.stop()
                }
        }
